package com.fanli.bigdata.mytest

import com.fanli.bigdata.db.StateRedisConnections
import kafka.serializer.StringDecoder
import org.apache.spark._
import org.apache.spark.rdd.RDD

import org.apache.log4j.{Level, Logger}
import java.sql.{Connection, DriverManager, ResultSet}

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.dstream.DStream

object VarnishDemo {
  Logger.getLogger("org").setLevel(Level.ERROR)
  val conf = new SparkConf().setAppName("MySpakDemo1").setMaster("local[*]")
  val sc = new SparkContext(conf)

  def main (args: Array[String]) {
//    val log = """111.19.32.224 - 104444908 [02/Nov/2016:17:43:54 +0800] GET http://fun.fanli.com/ HTTP/1.1 200 80 "http://m.fanli.com/taobao" "Fanli/5.4.2.11 (HUAWEI PLK-TL01H; Android 6.0; zh_CN; ID:2-104444908-67732835577999-12-0)" 0.011075258 miss pass 2168972125.2841786673.892491704 2168972125.2841786673.2733802942 21195b -"""
//    var varnish  =VarnishFun.getLogFunc(log)
//    println(varnish.day +"\t"+varnish.hour +"\t"+ varnish.uid)
//    varnishBatchMain()
    varnishSteamingMain()
  }

  def varnishBatchMain():Unit={
    var inputRDD = sc.textFile("file:///D:/varnish.txt")
    val log: RDD[(Realtime201611PUvKey,Realtime201611PUvValue)] = inputRDD.mapPartitions(VarnishFun.MatchLogBatchFunc(_))
    val res: Array[(Realtime201611PUvKey,Realtime201611PUvValue)] = log.collect()
    println(res.mkString(","))
    VarnishFun.logStat(res)
    sc.stop()
  }

  def varnishSteamingMain():Unit={
    var ssc = new StreamingContext(sc, Seconds(10))

    val kafkaParams = Map[String, String]("metadata.broker.list" -> "115.159.45.213:9092")
    val topics = "test02"
    val topicsSet = topics.split(",").toSet
    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    val log: DStream[(Realtime201611PUvKey,Realtime201611PUvValue)] = messages.mapPartitions(VarnishFun.MatchLogSteamingFunc(_))
    log.reduceByKeyAndWindow((v1:Realtime201611PUvValue,v2:Realtime201611PUvValue)=>{
      Realtime201611PUvValue(v1.us++v2.us,v1.count+v2.count)
    },Seconds(10),Seconds(10)).foreachRDD(rdd=>{
      val res: Array[(Realtime201611PUvKey,Realtime201611PUvValue)] = rdd.collect()
      println(res.mkString(","))
      VarnishFun.logStat(res)
    })
    ssc.start()
    ssc.awaitTermination()
  }
}
